package cn.doitedu.udfs;

import cn.doitedu.pojo.*;
import cn.doitedu.service.MatchService;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * MatchFunctionV4 是调用MatchServer，MatchFunctionV4相当于Controller
 */
public class MatchFunctionV4 extends KeyedProcessFunction<String, LogBean, MatchResult> {


    private RuleConditionV2 ruleCondition;

    private MatchService matchService;

    @Override
    public void open(Configuration parameters) throws Exception {
        //在TaskManager初始化subtask时，获取全局参数
        ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();

        //初始化MatchServer
        matchService = new MatchService();
        matchService.init(parameterTool, null, null);

//        //创建一个RuleCondition，以后使用广播状态动态广播获取
//        ruleCondition = new RuleConditionV2();
//        //模拟一个规则，以后是通过广播状态动态接入
//        //运营人员，事先定义好的规则
//        //触发事件参数
//        EventParam eventParam = new EventParam();
//        eventParam.setEventId("G");
//        HashMap<String, String> props = new HashMap<>();
//        props.put("p4", "v1");
//        props.put("p9", "v1");
//        eventParam.setProps(props);
//        ruleCondition.setTriggerEvent(eventParam);
//        //用户画像参数
//        HashMap<String, String> userProfiles = new HashMap<>();
//        userProfiles.put("tag1", "v2");
//        userProfiles.put("tag3", "v2");
//        ruleCondition.setUserProfiles(userProfiles);
//        //行为次数参数
//        CombineCondition eventCountCondition = new CombineCondition();
//
//        eventCountCondition.setStartTime(1649433600000L);
//        eventCountCondition.setEndTime(1649606399000L);
//        eventCountCondition.setCount(2);
//        eventCountCondition.setRegex("(1)");
//        eventCountCondition.setSql("select eventId from eagle_detail where deviceId = ? and timeStamp >= ? and timeStamp < ?");
//
//        EventParam eventCountParam = new EventParam();
//        //事件ID
//        eventCountParam.setEventId("G");
//        //行为参数
//        HashMap<String, String> eventCountProps = new HashMap<>();
//        eventCountProps.put("p4", "v1");
//        eventCountParam.setProps(eventCountProps);
//
//        eventCountCondition.setEvents(Arrays.asList(eventCountParam));
//
//        //行为序列参数
//        CombineCondition eventSeqCondition = new CombineCondition();
//
//        eventSeqCondition.setStartTime(1649433600000L);
//        eventSeqCondition.setEndTime(1649606399000L);
//        eventSeqCondition.setCount(1);
//        eventSeqCondition.setRegex("(.*?1.*?2)");
//        eventSeqCondition.setSql("select eventId from eagle_detail where deviceId = ? and timeStamp >= ? and timeStamp < ?");
//
//        //匹配的行为序列A -> H
//        EventParam eventParam1 = new EventParam();
//        eventParam1.setEventId("A");
//        EventParam eventParam2 = new EventParam();
//        eventParam2.setEventId("H");
//
//        eventSeqCondition.setEvents(Arrays.asList(eventParam1, eventParam2));
//
//        ruleCondition.setCombineConditions(Arrays.asList(eventCountCondition, eventSeqCondition));
    }

    @Override
    public void processElement(LogBean bean, Context ctx, Collector<MatchResult> out) throws Exception {
        //接收数据（LogBean）

        //调用Service，进行业务处理
        boolean flag = matchService.isMath(bean, ruleCondition);

        if (flag) {
            //返回结果
            out.collect(new MatchResult(bean.getDeviceId(), bean.getTimeStamp(), "rule001", System.currentTimeMillis()));
        }

    }

    @Override
    public void close() throws Exception {
        //释放资源
        matchService.close();
    }
}
